{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.streaming\n",
    "from pyspark.sql import SparkSession, types, functions\n",
    "from pyspark import SparkConf, SparkContext\n",
    "import sys\n",
    "from kafka import KafkaConsumer\n",
    "from pyspark.ml.classification import RandomForestClassifier, RandomForestClassificationModel\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "conf = SparkConf().setAppName('read stream')\n",
    "sc = SparkContext(conf=conf).getOrCreate()\n",
    "assert sys.version_info >= (3, 5)  # make sure we have Python 3.5+\n",
    "assert sc.version >= '2.2'  # make sure we have Spark 2.2+\n",
    "spark = SparkSession.builder.appName('nasa log').getOrCreate()\n",
    "\n",
    "#Defining the topic name where the data is published\n",
    "topic = 'test'\n",
    "\n",
    "#Consuming the message from Kafka message broker, topic = 'test'\n",
    "messages = spark.readStream.format('kafka') \\\n",
    "        .option('kafka.bootstrap.servers', 'localhost:9092') \\\n",
    "        .option('subscribe', topic).load()\n",
    "\n",
    "#Selecting the values in structured streaming\n",
    "message = messages.select(functions.decode(messages['value'],'UTF-8').alias('val'))\n",
    "msg = functions.split(message['val'],';')\n",
    "for i in range(0,64):\n",
    "    col = 'col_'+str(i)\n",
    "    message = message.withColumn(col, msg.getItem(i))\n",
    "\n",
    "def string_to_float(col):\n",
    "    return float(col)\n",
    "\n",
    "\n",
    "udf_float=functions.udf(string_to_float, returnType=types.FloatType())\n",
    "columns=message.schema.names\n",
    "for column in columns:\n",
    "        message=message.withColumn(column+'_f',udf_float(column)).drop(column)\n",
    "\n",
    "\n",
    "columns=message.schema.names\n",
    "temp=''\n",
    "for name in columns[2:]:\n",
    "    temp=temp+\"avg(abs(\"+name+\")),variance(abs(\"+name+\")),\"\n",
    "\n",
    "\n",
    "temp=temp.rstrip(\",\")\n",
    "\n",
    "message.createOrReplaceTempView(\"aggregate\")\n",
    "message=spark.sql('''select '''+temp+'''\n",
    "from aggregate ''')\n",
    "\n",
    "#Loading the Randomforest Classifier\n",
    "sameModel=RandomForestClassificationModel.load(\"/home/wolvorine/RFClassifier/stages/1_RandomForestClassifier_40ab83d4c98016a19d8d\")\n",
    "\n",
    "vecAssembler = VectorAssembler(inputCols=message.schema.names,outputCol=\"features\")\n",
    "message = vecAssembler.transform(message)\n",
    "\n",
    "#Applying the model\n",
    "message = sameModel.transform(message)\n",
    "stream = message.writeStream.format('console').outputMode('update').trigger(processingTime='3 seconds').start()\n",
    "\n",
    "stream.awaitTermination()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
